package org.meetunexpectedly.service.video.impl;

import org.meetunexpectedly.Lock.ILock;
import org.meetunexpectedly.constant.RedisConstant;
import org.meetunexpectedly.entity.user.User;
import org.meetunexpectedly.entity.video.Video;
import org.meetunexpectedly.entity.vo.Model;
import org.meetunexpectedly.entity.vo.UserModel;
import org.meetunexpectedly.exception.BaseException;
import org.meetunexpectedly.service.video.InterestPushService;
import org.meetunexpectedly.service.video.TypeService;
import org.meetunexpectedly.util.RedisCacheUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @Description: 兴趣推送服务实现类
 * @Author: pupil
 * @Date: 2024/08/13 下午 7:35
 */
@Service
public class InterestPushServiceImpl implements InterestPushService {
    // 最大标签数
    private static final int LABELS_SIZE_MAX = 100;

    // 获取的标签数
    private static final int VIDEO_LABEL_COUNT = 9;

    // 获取的视频数
    private static final int VIDEO_COUNT = 10;


    @Autowired
    private RedisCacheUtil redisCacheUtil;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private TypeService typeService;

    @Autowired
    private ILock lock;

    @Override
    @Async // 异步处理
    public void initUserModel(Long userId, List<String> labels) {
        String key = RedisConstant.USER_MODEL + userId;

        Map<Object, Object> userModelMap = new HashMap<>();
        if (!ObjectUtils.isEmpty(labels)) {
            int size = labels.size();
            // 将标签分为等分概率,不可能超过100个分类
            double probabilityValue = (double) LABELS_SIZE_MAX / size;
            for (String label : labels) {
                userModelMap.put(label, probabilityValue);
            }
        }

        // 删除旧缓存
        redisCacheUtil.del(key);

        // 设置兴趣推送模型缓存
        redisCacheUtil.hsSet(key, userModelMap);

        // 优化点 用户模型设置TTL，用户模型过期后根据性别推送
    }

    @Override
    @Async
    public void deleteUserModel(Long userId) {
        String key = RedisConstant.USER_MODEL + userId;
        redisTemplate.delete(key);
    }

    @Override
    @Async
    public void pushSystemStockIn(Video video) {
        // todo 还需要优化大key问题的解决方案
        // 获取标签
        List<String> labels = video.buildLabel();
        Long videoId = video.getId();
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (String label : labels) {
                connection.sAdd((RedisConstant.SYSTEM_STOCK + label).getBytes(), String.valueOf(videoId).getBytes());
            }
            return null;
        });
    }

    @Override
    @Async
    public void pushSystemTypeStockIn(Video video) {
        // 使用分片key和id生成器来解决大key问题
        Long typeId = video.getTypeId();
        // 判断分类的分片自增id是否存在
        boolean keyExists = redisCacheUtil.isKeyExists(RedisConstant.SHARD_TYPE_ID + typeId);

        // 获取分片自增id
        int shardNumber = 1;
        if (!keyExists) {
            // 分片自增id不存在，进行添加
            redisCacheUtil.set(RedisConstant.SHARD_TYPE_ID + typeId, 1);
        } else {
            // 分片自增id存在，从redis获取最新的
            Object o = redisCacheUtil.get(RedisConstant.SHARD_TYPE_ID + typeId);
            shardNumber = (int) o;
        }

        String key = RedisConstant.SYSTEM_TYPE_STOCK + typeId + ":" + shardNumber;
        Long maxDataCount = RedisConstant.MAX_DATA_COUNT;
        Long setSize = redisCacheUtil.getSetSize(key);
        // 判断当前分片的数据量是否需要添加新的分片
        if (setSize >= maxDataCount) {
            shardNumber += 1;
            redisCacheUtil.set(RedisConstant.SHARD_TYPE_ID + typeId, shardNumber);
            key = RedisConstant.SYSTEM_TYPE_STOCK + typeId + ":" + shardNumber;
        }

        redisCacheUtil.sSet(key, video.getId());
    }

    @Override
    @Async
    public void deleteSystemStockIn(Video video) {
        // todo 还需要优化大key问题的解决方案
        List<String> labels = video.buildLabel();
        Long videoId = video.getId();
       try {
           redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
               for (String label : labels) {
                   connection.sRem((RedisConstant.SYSTEM_STOCK + label).getBytes(), String.valueOf(videoId).getBytes());
               }
               return null;
           });
       }catch (Exception e){
           e.printStackTrace();
       }
    }

    @Override
    @Async
    public void deleteSystemTypeStockIn(Video video) {
        // 使用分片key和id生成器来解决大key问题
        Long typeId = video.getTypeId();
        // 判断分类的分片自增id是否存在
        boolean keyExists = redisCacheUtil.isKeyExists(RedisConstant.SHARD_TYPE_ID + typeId);
        if (!keyExists) {
            throw new BaseException("分类key不存在");
        }

        // 分片自增id存在，从redis获取最新的
        Object o = redisCacheUtil.get(RedisConstant.SHARD_TYPE_ID + typeId);
        int shardNumber = o != null ? (int) o : 1;

        boolean videoDeleted = false;
        int tempShardNumber = shardNumber;
        while (tempShardNumber > 0) {
            long count = redisCacheUtil.setRemove(RedisConstant.SYSTEM_TYPE_STOCK + typeId + ":" + tempShardNumber, video.getId());
            if (count == 1) {
                videoDeleted = true;
                break;
            }
            tempShardNumber--;
        }

        if (!videoDeleted) {
            // 抛出异常
            throw new BaseException("视频未找到，删除失败");
        }

        // todo 需要优化数据合并的处理方式
        Long setSize = redisCacheUtil.getSetSize(RedisConstant.SYSTEM_TYPE_STOCK + typeId + ":" + tempShardNumber);
        if (setSize <= 15) {
            try {
                boolean acquired = lock.tryLock(10L);
                if (!acquired) {
                    throw new BaseException("无法获取分片合并的锁");
                }
                for (; tempShardNumber < shardNumber; tempShardNumber++) {
                    String key1 = RedisConstant.SYSTEM_TYPE_STOCK + typeId + ":" + tempShardNumber;
                    String key2 = RedisConstant.SYSTEM_TYPE_STOCK + typeId + ":" + tempShardNumber + 1;
                    // 获取下一个分片数据
                    Set<Object> set2 = redisCacheUtil.sGet(key2);
                    if (set2 != null && !set2.isEmpty()) {
                        redisCacheUtil.sSet(key1, set2.toArray());
                        // 清空并保留 key2
                        redisTemplate.execute((RedisCallback<Object>) connection -> {
                            connection.multi();  // 开启事务
                            connection.del(key2.getBytes());  // 删除所有元素
                            connection.sAdd(key2.getBytes(), new byte[][]{});  // 保留 key
                            connection.exec();  // 提交事务
                            return null;
                        });
                    }
                }
            } finally {
                lock.unlock();
            }

        }
    }

    @Override
    @Async
    public void updateUserModel(UserModel userModel) {
        Long userId = userModel.getUserId();
        // 游客不用管
        if (userId != null) {
            List<Model> models = userModel.getModels();
            String key = RedisConstant.USER_MODEL + userId;
            // 从缓存中获取用户模型：记录标签和标签对应的概率
            Map<Object, Object> userModelMap = redisCacheUtil.hsGet(key);

            // 没有缓存
            if (ObjectUtils.isEmpty(userModelMap)) {
                userModelMap = new HashMap<>();
            }
            double maxProbability = Double.MIN_VALUE;
            for (Model model : models) {
                // 修改用户模型
                if (userModelMap.containsKey(model.getLabel())) {
                    double newProbability = Double.parseDouble(userModelMap.get(model.getLabel()).toString()) + model.getScore();
                    if(newProbability > maxProbability) maxProbability = newProbability;
                    // 增加标签概率(旧概率+增加的概率)
                    userModelMap.put(model.getLabel(), newProbability);
                    // 标签的概率
                    Object o = userModelMap.get(model.getLabel());
                    if (o == null || Double.parseDouble(o.toString()) < 0.0) {
                        userModelMap.remove(model.getLabel());
                    }

                } else {
                    if (model.getLabel() != null) {
                        // 新标签直接存入
                        userModelMap.put(model.getLabel(), model.getScore());
                    }
                }
            }

            // 当标签中最大概率大于等于50时进行数据缩放
            if(maxProbability >= 50.0){
                int labelSize = userModelMap.keySet().size();
                for (Object o : userModelMap.keySet()) {
                    // 每个标签概率同等加上标签数，再同等除以标签数  防止数据膨胀(需要优化)
//                    userModelMap.put(o, (Double.parseDouble(userModelMap.get(o).toString()) + labelSize) / labelSize);
                    // 将概率减半以防止数据膨胀
                    userModelMap.put(o, (Double.parseDouble(userModelMap.get(o).toString()) / 2));
                }
            }

            // 更新用户模型
            redisCacheUtil.hsSet(key, userModelMap);
        }

    }

    @Override
    public Collection<Long> listVideoIdsByUserModel(User user) {
        // 创建结果集
        // 随机获取VIDEO_COUNT个视频
        Set<Long> videoIds = new HashSet<>(VIDEO_COUNT);

        if (user != null) {
            // 用户
            Map<Object, Object> userModelMap = redisCacheUtil.hsGet(RedisConstant.USER_MODEL + user.getId());
            if (!ObjectUtils.isEmpty(userModelMap)) {
                String[] probabilityArray = initProbabilityArray(userModelMap);
                Boolean sex = user.getSex();

                Random random = new Random();
                ArrayList<String> labelNames = new ArrayList<>();

                // 随机获取VIDEO_LABEL_COUNT个标签名
                for (int i = 1; i <= VIDEO_LABEL_COUNT; i++) {
                    String labelName = probabilityArray[random.nextInt(probabilityArray.length)];
                    labelNames.add(labelName);
                }
                String sysKey = RedisConstant.SYSTEM_STOCK;
                List<Object> videoIdsList = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                    for (String labelName : labelNames) {
                        String key = sysKey + labelName;
                        // 随机返回一个
                        connection.sRandMember(key.getBytes());
                    }
                    return null;
                });
                // 因为videoIdsList可能会有重复的videId存在，所以要转换成set集合进行去重
                Set<Long> ids = videoIdsList.stream().filter(id -> !ObjectUtils.isEmpty(id)).map(id -> Long.parseLong(id.toString())).collect(Collectors.toSet());

                // 去重，去除用户观看过的视频
                String historyKey = RedisConstant.HISTORY_VIDEO;
                Long userId = user.getId();

                // 获取用户观看过的视频
                List<Object> duplicateIds = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                    for (Long id : ids) {
                        connection.get((historyKey + id + ":" + userId).getBytes());
                    }
                    return null;
                });
                duplicateIds = duplicateIds.stream().filter(o -> !ObjectUtils.isEmpty(o)).map(id -> Long.parseLong(id.toString())).collect(Collectors.toList());
                // 执行去重
                if (!ObjectUtils.isEmpty(duplicateIds)) {
                    for (Object duplicateId : duplicateIds) {
                        duplicateId = Long.parseLong(duplicateId.toString());
                        if (ids.contains(duplicateId)) {
                            ids.remove(duplicateId);
                        }
                    }
                }
                videoIds.addAll(ids);
                // 根据性别随机获取一个视频
                Long videoId = randomVideoId(sex);
                if (videoId != null) {
                    videoIdsList.add(videoId);
                }
                System.out.println();
            }

        } else {
            // 游客
            // 随机获取10个标签
            List<String> labels = typeService.randomLabels(VIDEO_COUNT);
            List<String> labelNames = new ArrayList<>();
            for (String label : labels) {
                labelNames.add(RedisConstant.SYSTEM_STOCK + label);
            }
            Collections.shuffle(labelNames);
            // 获取videoId，已经进行去空处理
            List<Object> videoIdsList = redisCacheUtil.sRandom(labelNames);
            if (!ObjectUtils.isEmpty(videoIdsList)) {
                Set<Long> ids = videoIdsList.stream().map(id -> Long.parseLong(id.toString())).collect(Collectors.toSet());
                videoIds.addAll(ids);
            }
        }

        return videoIds;
    }

    @Override
    public Collection<Long> listVideoIdByTypeId(Long typeId) {
        // 随机推送15个视频
        Object o = redisCacheUtil.get(RedisConstant.SHARD_TYPE_ID + typeId);
        int shardNumber = o != null ? (int) o : 1;
        Random random = new Random();
        shardNumber = random.nextInt(shardNumber) + 1;
        List<Object> videoIds = redisTemplate.opsForSet().randomMembers(RedisConstant.SYSTEM_TYPE_STOCK + typeId + ":" + shardNumber, 15);

        if (ObjectUtils.isEmpty(videoIds)) {
            return Collections.EMPTY_LIST;
        }

        final HashSet<Long> result = new HashSet<>();
        for (Object videoId : videoIds) {
            if (videoId != null) {
                result.add(Long.parseLong(videoId.toString()));
            }
        }
        return result;
    }

    @Override
    public Collection<Long> listVideoIdByLabels(List<String> labelNames) {

        List<String> labelKeys = new ArrayList<>();
        for (String labelName : labelNames) {
            labelKeys.add(RedisConstant.SYSTEM_STOCK + labelName);
        }

        // 创建结果集
        Set<Long> videoIds = new HashSet<>();
        // 已经进行去空处理
        List<Object> list = redisCacheUtil.sRandom(labelKeys);
        if (!ObjectUtils.isEmpty(list)) {
            videoIds = list.stream().map(id -> Long.parseLong(id.toString())).collect(Collectors.toSet());
        }

        return videoIds;
    }

    /**
     * 根据性别随获取一个视频
     * 男：美女，游戏   女：
     *
     * @param sex
     * @return
     */
    private Long randomVideoId(Boolean sex) {
        String[] labels = sex ? new String[]{"美女", "音乐","知识"} : new String[]{"宠物", "音乐","穿搭"};
        Random random = new Random();
        String label = labels[random.nextInt(labels.length)];
        Object o = redisCacheUtil.sRandom(RedisConstant.SYSTEM_STOCK + label);
        if (!ObjectUtils.isEmpty(o)) {
            long videoId = Long.parseLong(o.toString());
            return videoId;
        }
        return null;
    }

    /**
     * 初始化概率数组 -> 保存的元素是标签
     * 同一个标签根据概率存储对应的个数，通过数组中标签的个数替换概率
     *
     * @param modelMap
     * @return
     */
    private String[] initProbabilityArray(Map<Object, Object> modelMap) {
        // key: 标签  value：概率
        Map<String, Integer> probabilityMap = new HashMap<>();
        int size = modelMap.size();
        AtomicInteger counter = new AtomicInteger();
        modelMap.forEach((k, v) -> {
            // 防止结果为0,每个同等加上标签数
            int probability = (((Double) v).intValue() + size )/ size;
            counter.getAndAdd(probability);
            probabilityMap.put(k.toString(), probability);
        });

        // 结果集
        String[] probabilityArray = new String[counter.get()];

        AtomicInteger index = new AtomicInteger();
        probabilityMap.forEach((label, probability) -> {
            int i = index.get();
            int limit = i + probability;
            while (i < limit) {
                probabilityArray[i++] = label;
            }
            index.set(limit);
        });
        return probabilityArray;
    }
}
